package avicit.bdp.dcs.json.service.builder;

import avicit.bdp.common.utils.CheckParaUtils;
import avicit.bdp.dcs.datasource.dto.JobDatasource;
import avicit.bdp.dcs.datasource.service.DataSourceUtils;
import avicit.bdp.dcs.job.dto.JobDTO;
import avicit.bdp.dcs.job.service.DcsJobService;
import avicit.bdp.dcs.json.service.reader.StreamReaderBuilder;
import avicit.bdp.dcs.json.service.util.BuildJsonUtil;
import avicit.bdp.dcs.json.service.writer.StreamWriterBuilder;
import avicit.bdp.dcs.plugin.BaseDataxParam;
import avicit.bdp.dcs.task.dto.TaskDTO;
import avicit.bdp.dcs.task.service.TaskService;
import avicit.bdp.dcs.utils.DcsErrorCode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
public class StreamJsonBuilder {

  private static final Logger logger = LoggerFactory.getLogger(StreamJsonBuilder.class);

  @Autowired private TaskService taskService;
  @Autowired private DcsJobService jobService;
  @Autowired private BuildJsonUtil buildJsonUtil;
  @Autowired private DataSourceUtils dataSourceUtils;
  @Autowired private StreamWriterBuilder streamWriterBuilder;
  @Autowired private StreamReaderBuilder streamReaderBuilder;

  /**
   * 流式采集
   *
   * @param taskId 任务ID
   * @return datax json
   * @throws Exception build json exception
   */
  public String buildDataxJsonByStream(String taskId) throws Exception {
    // 1、参数合法性校验
    TaskDTO taskDTO = taskService.selectByPrimaryKey(taskId);
    CheckParaUtils.checkDbObjectValidity(
        taskDTO, logger, String.format(DcsErrorCode.TASK_NULL.getDesc(), taskId));
    JobDTO jobDTO = jobService.selectByPrimaryKey(taskDTO.getJobId());
    CheckParaUtils.checkDbObjectValidity(
        jobDTO, logger, String.format(DcsErrorCode.JOB_NULL.getDesc(), taskDTO.getJobId()));

    JobDatasource readerDatasource = dataSourceUtils.convertDatasource(jobDTO.getSrcDatasourceId());
    JobDatasource writerDatasource = dataSourceUtils.convertDatasource(jobDTO.getDstDatasourceId());

    // 2、构建Reader
    BaseDataxParam readerParam =
        streamReaderBuilder.buildReaderByStream(taskDTO, readerDatasource, jobDTO);

    // 3、构建Writer
    BaseDataxParam writerParam =
        streamWriterBuilder.buildWriterByStream(taskDTO, jobDTO, writerDatasource);

    return buildJsonUtil.buildJobJson(
        jobDTO.getSrcDatasourceId(),
        jobDTO.getDstDatasourceId(),
        readerParam,
        writerParam,
        null,
        jobDTO,
        taskDTO);
  }
}
